package felix.actor.core;

import felix.actor.core.msg.TActorMsg;
import felix.actor.core.util.ThingsRouterThreadFactory;
import lombok.Data;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Predicate;
import java.util.stream.Collectors;

@Slf4j
@Data
public class DefaultTActorSystem implements TActorSystem {

    private final ConcurrentMap<String, Dispatcher> dispatchers = new ConcurrentHashMap<>();
    private final ConcurrentMap<TActorId, TActorMailbox> actors = new ConcurrentHashMap<>();
    private final ConcurrentMap<TActorId, ReentrantLock> actorCreationLocks = new ConcurrentHashMap<>();
    private final ConcurrentMap<TActorId, Set<TActorId>> parentChildMap = new ConcurrentHashMap<>();

    @Getter
    private final TActorSystemSettings settings;
    @Getter
    private final ScheduledExecutorService scheduler;

    public DefaultTActorSystem(TActorSystemSettings settings) {
        this.settings = settings;
        this.scheduler = Executors.newScheduledThreadPool(settings.getSchedulerPoolSize(), ThingsRouterThreadFactory.forName("actor-system-scheduler"));
    }

    @Override
    public void createDispatcher(String dispatcherId, ExecutorService executor) {
        Dispatcher current = dispatchers.putIfAbsent(dispatcherId, new Dispatcher(dispatcherId, executor));
        if (current != null) {
            throw new RuntimeException("Dispatcher with id [" + dispatcherId + "] is already registered!");
        }
    }

    @Override
    public void destroyDispatcher(String dispatcherId) {
        Dispatcher dispatcher = dispatchers.remove(dispatcherId);
        if (dispatcher != null) {
            dispatcher.getExecutor().shutdownNow();
        } else {
            throw new RuntimeException("Dispatcher with id [" + dispatcherId + "] is not registered!");
        }
    }

    @Override
    public TActorRef getActor(TActorId actorId) {
        return actors.get(actorId);
    }

    @Override
    public TActorRef createRootActor(String dispatcherId, TActorCreator creator) {
        return createActor(dispatcherId, creator, null);
    }

    @Override
    public TActorRef createChildActor(String dispatcherId, TActorCreator creator, TActorId parent) {
        return createActor(dispatcherId, creator, parent);
    }

    private TActorRef createActor(String dispatcherId, TActorCreator creator, TActorId parent) {
        Dispatcher dispatcher = dispatchers.get(dispatcherId);
        if (dispatcher == null) {
            log.warn("Dispatcher with id [{}] is not registered!", dispatcherId);
            throw new RuntimeException("Dispatcher with id [" + dispatcherId + "] is not registered!");
        }

        TActorId actorId = creator.createActorId();
        TActorMailbox actorMailbox = actors.get(actorId);
        if (actorMailbox != null) {
            log.debug("Actor with id [{}] is already registered!", actorId);
        } else {
            Lock actorCreationLock = actorCreationLocks.computeIfAbsent(actorId, id -> new ReentrantLock());
            actorCreationLock.lock();
            try {
                actorMailbox = actors.get(actorId);
                if (actorMailbox == null) {
                    log.debug("Creating actor with id [{}]!", actorId);
                    TActor actor = creator.createActor();
                    TActorRef parentRef = null;
                    if (parent != null) {
                        parentRef = getActor(parent);
                        if (parentRef == null) {
                            throw new TActorNotRegisteredException(parent, "Parent Actor with id [" + parent + "] is not registered!");
                        }
                    }
                    TActorMailbox mailbox = new TActorMailbox(this, settings, actorId, parentRef, actor, dispatcher);
                    actors.put(actorId, mailbox);
                    mailbox.initActor();
                    actorMailbox = mailbox;
                    if (parent != null) {
                        parentChildMap.computeIfAbsent(parent, id -> ConcurrentHashMap.newKeySet()).add(actorId);
                    }
                } else {
                    log.debug("Actor with id [{}] is already registered!", actorId);
                }
            } finally {
                actorCreationLock.unlock();
                actorCreationLocks.remove(actorId);
            }
        }
        return actorMailbox;
    }

    @Override
    public void tellWithHighPriority(TActorId target, TActorMsg actorMsg) {
        tell(target, actorMsg, true);
    }

    @Override
    public void tell(TActorId target, TActorMsg actorMsg) {
        tell(target, actorMsg, false);
    }

    private void tell(TActorId target, TActorMsg actorMsg, boolean highPriority) {
        TActorMailbox mailbox = actors.get(target);
        if (mailbox == null) {
            throw new TActorNotRegisteredException(target, "Actor with id [" + target + "] is not registered!");
        }
        if (highPriority) {
            mailbox.tellWithHighPriority(actorMsg);
        } else {
            mailbox.tell(actorMsg);
        }
    }


    @Override
    public void broadcastToChildren(TActorId parent, TActorMsg msg) {
        broadcastToChildren(parent, id -> true, msg);
    }

    @Override
    public void broadcastToChildren(TActorId parent, Predicate<TActorId> childFilter, TActorMsg msg) {
        Set<TActorId> children = parentChildMap.get(parent);
        if (children != null) {
            children.stream().filter(childFilter).forEach(id -> tell(id, msg));
        }
    }

    @Override
    public List<TActorId> filterChildren(TActorId parent, Predicate<TActorId> childFilter) {
        Set<TActorId> children = parentChildMap.get(parent);
        if (children != null) {
            return children.stream().filter(childFilter).collect(Collectors.toList());
        } else {
            return Collections.emptyList();
        }
    }

    @Override
    public void stop(TActorRef actorRef) {
        stop(actorRef.getActorId());
    }

    @Override
    public void stop(TActorId actorId) {
        Set<TActorId> children = parentChildMap.remove(actorId);
        if (children != null) {
            for (TActorId child : children) {
                stop(child);
            }
        }
        TActorMailbox mailbox = actors.remove(actorId);
        if (mailbox != null) {
            mailbox.destroy();
        }
    }

    @Override
    public void stop() {
        dispatchers.values().forEach(dispatcher -> {
            dispatcher.getExecutor().shutdown();
            try {
                dispatcher.getExecutor().awaitTermination(3, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                log.warn("[{}] Failed to stop dispatcher", dispatcher.getDispatcherId(), e);
            }
        });
        if (scheduler != null) {
            scheduler.shutdownNow();
        }
        actors.clear();
    }

}
